package com.suning.sawp.test.bi;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.Properties;


import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * 单例模式
 * @author sczhou
 */
public class KafkaProducer {
	private kafka.javaapi.producer.Producer<String, String> producer;

	private String topic;
	private Properties props = new Properties();
	private static KafkaProducer uniqueInstance = null;
	private int messageNo = 28261;

	public KafkaProducer() {
		
	    String kafkaZkServer = "10.27.25.161:9092,10.27.25.163:9093";
		//此处配置的是kafka的端口
		props.put("metadata.broker.list", kafkaZkServer);
		//配置value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        //配置key的序列化类
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        
        this.topic = "mbd_comm_sendorderdetail";//"bi-dfp-oms-detail";
        
		producer = new kafka.javaapi.producer.Producer<String, String>(
				new ProducerConfig(props));
		
	}
	/**
	 * 取单例模式实例
	 * @return
	 */
	public final static KafkaProducer getInstance() {
		if (uniqueInstance == null) {
			uniqueInstance = new KafkaProducer();
		}
		return uniqueInstance;
	}

	public void produce(String message) {
		String messageStr = String.valueOf(messageNo);
		producer.send(new KeyedMessage<String, String>(topic, messageStr, message));
		messageNo++;
	}
	
	public static void main(String[] args){
	    KafkaProducer p = new KafkaProducer();
	    String orderJson = null;
	    String line = null;
	    try {
            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(
                    "D://oms_order_test.log"), "UTF-8"));
            while (null != (line = br.readLine())) {
                orderJson += line;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
	    p.produce(orderJson);
	    System.err.println("over");
	}

}
